package cn.lokei.gateway.handler.message;

import cn.lokei.gateway.service.ChannelSupervise;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.channel.group.ChannelGroup;

public class ReceiveHandler {

  private ChannelSupervise channelSupervise = ChannelSupervise.getBean(ChannelSupervise.class);

  public void receiveMessage(MqttMessage mqttMessage) {
    MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
    String topic = mqttPublishMessage.variableHeader().topicName();
    ChannelGroup channelGroup = channelSupervise.subChannels(topic);
    if ((!(channelGroup == null)) && channelGroup.size() > 0) {
      MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false,
          mqttPublishMessage.fixedHeader().qosLevel(), false, mqttPublishMessage.fixedHeader().remainingLength());
      // 构建返回报文， 可变报头
      MqttPublishVariableHeader mqttMessageIdVariableHeader = new MqttPublishVariableHeader(topic,
          mqttPublishMessage.variableHeader().packetId());
      MqttPublishMessage mqttMessagePublish = new MqttPublishMessage(mqttFixedHeader, mqttMessageIdVariableHeader,
          mqttPublishMessage.payload());
      channelGroup.writeAndFlush(mqttMessagePublish);
    }
  }
}
